from absl import app, flags
import collections
import datetime

from coin.tool.strat_monitor.app.alarm_strategy_log import StrategyLogAlarm
from coin.tool.strat_monitor.handlers.strategy_alarm_handler import *
from coin.proto.coin_executor_pb2 import PassiveExecutorConfig
from coin.proto.coin_request_pb2 import ProcessInfoProto, AccountRequestProto
from coin.proto.coin_order_gateway_pb2 import OrderGatewayLog, OrderEvent
from coin.proto.coin_query_pb2 import AccountPosition, ProductPosition, CurrencyBalance

FLAGS = flags.FLAGS

Record = collections.namedtuple('Record', ['topic', 'value', 'timestamp', 'type'])

MACHINE = 'local-machine'
STRAT_NAME = 'dmm_okex_binance_usdt_3_1'
TOPIC = 'STRAT_%s_%s' % (MACHINE, STRAT_NAME)
test_account = {
  'market_type': 'Spot',
  'exchange': 'Okex',
  'owner': 'prestoinvt24'
}


pass_executor_config = PassiveExecutorConfig(
  symbol='BTC-USDT',
  lot_size=10,
  max_pos=30,
  min_pos=0
)

class FakeRecord():
  def __init__(self, topic=TOPIC):
    self.topic = topic

  def get_executor_config(self, ts):
    strategy_log = StrategyLog(
      type=StrategyLog.PASS_EXECUTOR_CONFIG,
      pass_executor_config=pass_executor_config
    )
    return Record(topic=self.topic,
                  value=strategy_log,
                  timestamp=ts,
                  type='config')

  def get_process(self, ts, pid=None):
    pid = pid or 100
    strategy_log = StrategyLog(
      process=ProcessInfoProto(machine=MACHINE, pid=pid)
    )
    return Record(topic=self.topic,
                  value=strategy_log,
                  timestamp=ts,
                  type='process')

  def get_order_event(self, ts, order_type):
    order_event = OrderEvent(type=order_type, symbol='BTC-USDT')
    og_log = OrderGatewayLog(type=OrderGatewayLog.ORDER_EVENT, event=order_event)
    strategy_log = StrategyLog(
      type=StrategyLog.OG_LOG,
      og_log=og_log
    )
    return Record(topic=self.topic,
                  value=strategy_log,
                  timestamp=ts,
                  type=order_type)

  def get_account_request(self,ts, abnormal=False):
    if abnormal is True:
      end_str = '_abnormal'
    else:
      end_str = ''
    account_request = AccountRequestProto(
      market_type=test_account['market_type']+end_str,
      exchange=test_account['exchange']+end_str,
      owner=test_account['owner']+end_str,
    )
    og_log = OrderGatewayLog(account_request=account_request)
    strategy_log = StrategyLog(
      type=StrategyLog.OG_LOG,
      og_log=og_log
    )
    return Record(topic=self.topic,
                  value=strategy_log,
                  timestamp=ts,
                  type='account_request')

  def get_pnl_balance(self, ts):
    pnl_balance = CurrencyBalance()
    strategy_log = StrategyLog(pnl_balance=pnl_balance)
    return Record(topic=self.topic,
                  value=strategy_log,
                  timestamp=ts,
                  type='pnl_balance')

  def get_position(self, ts, abnormal_stack=False, abnormal_symbol=False):
    if abnormal_stack is True:
      net_pos = 10000
    else:
      net_pos = 10
    each_pos = [ProductPosition(symbol='BTC-USDT', net_position=net_pos)]
    if abnormal_symbol:
      each_pos.append(ProductPosition(symbol='ETH-USDT', net_position=net_pos))
    position = AccountPosition(each_position=each_pos)
    account_request = AccountRequestProto(
      market_type=test_account['market_type'],
      exchange=test_account['exchange'],
      owner=test_account['owner'],
    )
    og_log = OrderGatewayLog(position=position, account_request=account_request, type=OrderGatewayLog.POSITION)
    strategy_log = StrategyLog(
      type=StrategyLog.OG_LOG,
      og_log=og_log
    )
    return Record(topic=self.topic,
                  value=strategy_log,
                  timestamp=ts,
                  type='position')

  def gen_kafak_record(self, start_ts:int, end_ts:int):
    record_list = []
    step = 200
    iter_ts = (end_ts - start_ts) // 200
    for i in range(1, step):
      ts = start_ts + i * iter_ts
      if i % 20 == 0:
        record_list.append(self.get_executor_config(ts))
      if i % 2 == 0:
        record_list.append(self.get_order_event(ts, OrderEvent.ORDER_SUBMITTED))
      if i % 5 == 0:
        record_list.append(self.get_order_event(ts, OrderEvent.ORDER_REJECTED))
      if i % 10 == 0:
        record_list.append(self.get_account_request(ts))
      if i % 20 == 0:
        record_list.append(self.get_process(ts))
      if i % 30 == 0:
        record_list.append(self.get_position(ts))
        record_list.append(self.get_pnl_balance(ts))
    return record_list

  def convert_to_multi_process(self, record_list):
    last_ts = record_list[-1].timestamp
    record_list.extend([self.get_process(last_ts), self.get_process(last_ts, pid=666), self.get_process(last_ts), self.get_process(last_ts)])
    return record_list

  def convert_to_no_log(self, record_list):
    return []

  def convert_to_no_pnlbalance(self, record_list):
    return [record for record in record_list if record.type != 'pnl_balance']

  def convert_to_multi_machine(self, record_list):
    last_ts = record_list[-1].timestamp
    record = self.get_order_event(last_ts, None)
    record = record._replace(topic='STRAT_machine111_%s' % STRAT_NAME)
    record_list.append(record)
    return record_list

  def convert_to_abnormal_stack(self, record_list):
    last_ts = record_list[-1].timestamp
    record = self.get_position(last_ts, abnormal_stack=True)
    record_list.insert(-1, record)
    return record_list

  def convert_to_abnormal_position(self, record_list):
    last_ts = record_list[-1].timestamp
    record = self.get_position(last_ts, abnormal_symbol=True)
    record_list.insert(-1, record)
    return record_list

  def convert_to_ordererror(self, record_list):
    last_ts = record_list[-1].timestamp
    diff = len([r for r in record_list if r.type==OrderEvent.ORDER_SUBMITTED]) \
           - len([r for r in record_list if r.type==OrderEvent.ORDER_REJECTED])
    if diff > 0:
      record_list.extend([self.get_order_event(last_ts, OrderEvent.ORDER_REJECTED)] * diff)
    return record_list

def alarm_warm_up(alarm, fake_record):
  end_ts = get_timestamp() - 11 * 60 * 1e+9
  start_ts = int(end_ts - 5 * 60 * 1e+9)
  record_list = fake_record.gen_kafak_record(start_ts, end_ts)
  alarm.update_all_timestamp(end_ts)
  for record in record_list:
    topic = record.topic
    ts = int(record.timestamp)
    strategy = topic.split('_', 2)[-1]
    pb = record.value
    alarm._handlers[strategy].handle(ts, topic, pb)
  return alarm

def split_record_list(record_list, diff=30*1e+9):
  start_ts = record_list[0].timestamp
  res = []
  temp_record_list = []
  for record in record_list:
    if record.timestamp < start_ts + diff:
      temp_record_list.append(record)
    else:
      res.append(temp_record_list)
      temp_record_list = [record]
      start_ts += diff
  res.append(temp_record_list)
  return res


def main(_):
  alarm = StrategyLogAlarm(kafka_config_filename=FLAGS.kafka_config_filename)
  alarm._handlers[STRAT_NAME] = StrategyAlarmHandler(strategy=STRAT_NAME)
  alarm.update_all_application_human_view()
  alarm.update_all_retired_strategy()
  fake_record = FakeRecord(topic=TOPIC)
  alarm = alarm_warm_up(alarm, fake_record)
  end_ts = get_timestamp()
  start_ts = int(end_ts - 11 * 60 * 1e+9)
  record_list = fake_record.gen_kafak_record(start_ts, end_ts)
  record_list = fake_record.convert_to_abnormal_stack(record_list)
  record_list = fake_record.convert_to_ordererror(record_list)
  record_list = fake_record.convert_to_multi_machine(record_list)
  record_list = fake_record.convert_to_no_pnlbalance(record_list)
  record_list = fake_record.convert_to_multi_process(record_list)
  record_list = fake_record.convert_to_abnormal_position(record_list)
  record_lists = split_record_list(record_list)
  for i, record_list in enumerate(record_lists):
    alarm.update_all_timestamp(record_list[-1].timestamp)
    for record in record_list:
      topic = record.topic
      ts = int(record.timestamp)
      strategy = topic.split('_', 2)[-1]
      pb = record.value
      alarm._handlers[strategy].handle(ts, topic, pb)
    alarm._logger.info("end receive at %s" % datetime.datetime.utcfromtimestamp(record.timestamp/1e+9))
    alarm.report_abnormals()
  return


if __name__ == '__main__':
  logging.basicConfig(level='INFO', format='%(levelname)8s %(asctime)s %(message)s')
  flags.DEFINE_string('kafka_config_filename',
                      '../../coin_deploy/support_monitor/config/kafka_config.json', 'kafka config')
  flags.DEFINE_string('memcached_config',
                      "../../coin_deploy/support_accounting/db_config/memcached_config_prod.json",
                      'path of database config.')
  flags.DEFINE_string('slack_channel', None, 'channel to receive alarm msg')
  app.run(main)




